package com.behelpful.demo.search.event;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.annotation.TableName;
import com.behelpful.common.core.exceptions.IRuntimeException;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.behelpful.demo.search.model.ISearchModel;
import com.behelpful.demo.search.service.ISearchService;
import com.behelpful.demo.search.service.SqlSchemaColumnService;
import org.elasticsearch.action.ActionListener;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.io.Serializable;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 具体的 BinLog 事件增删改事件处理的逻辑代码和公共的变量管理存储
 *
 * @author Alay
 * @date 2021-08-05 00:19
 * @project behelpful
 */
public abstract class AbsBinLogEvent<T extends ISearchModel> implements BinaryLogClient.EventListener {

    /**
     * Map< tableId, MySQL表名>
     */
    protected volatile static Map<Long, String> tableFullNameMap = new HashMap<>(1 << 3);
    /**
     * Map< tableId, MySQL表对应 Java 实体类的类对象 Class>
     */
    protected volatile static Map<String, AbsBinLogEvent> eventInstanceMap = new HashMap<>(1 << 4);

    protected ISearchService<T> baseEsService;

    @Autowired
    protected ActionListener updateListener;
    @Autowired
    protected ActionListener indexListener;
    @Autowired
    protected ActionListener deleteListener;

    @Autowired
    protected SqlSchemaColumnService sqlSchemaColumnService;

    /**
     * 初始化需要处理的表的事件处理类
     * 初始化变量，及注入
     *
     * @throws IRuntimeException
     */
    @PostConstruct
    protected abstract void init() throws IRuntimeException;

    @Override
    public void onEvent(Event event) {
        EventData eventData = event.getData();
        /**
         * 修改操作
         */
        if (eventData instanceof UpdateRowsEventData) {
            T entity = this.updateEvent(eventData);
            // 同步 ES 的操作
            T esEntity = baseEsService.selectById(entity.getEsId());
            if (null == esEntity) {
                baseEsService.saveEntityAsy(entity, indexListener);
            } else {
                baseEsService.updateByIdAsy(entity, updateListener);
            }
            return;
        }

        /**
         * 写入操作
         */
        if (eventData instanceof WriteRowsEventData) {
            T entity = this.saveEvent(eventData);
            // 写入数据
            baseEsService.saveEntityAsy(entity, indexListener);
            return;
        }
        /**
         * 删除操作
         */
        if (eventData instanceof DeleteRowsEventData) {
            T entity = this.deleteEvent(eventData);
            // 删除 ES 中数据
            baseEsService.deleteByIdAsy(entity.getEsId(), deleteListener);
            return;
        }
    }

    protected T saveEvent(EventData eventData) {
        WriteRowsEventData writeRowsEventData = (WriteRowsEventData) eventData;
        List<Serializable[]> rows = writeRowsEventData.getRows();
        T entity = this.rowsToEntity(rows, writeRowsEventData.getTableId());
        return entity;
    }

    protected T updateEvent(EventData eventData) {
        UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) eventData;
        // rows 每一个 Entry 是条记录,其中 Key 为修改前的记录，Value 为修改后的新的记录
        List<Map.Entry<Serializable[], Serializable[]>> rows = updateRowsEventData.getRows();
        // 获取修改后的新的值
        List<Serializable[]> newValues = rows.stream().map(entry -> entry.getValue()).collect(Collectors.toList());
        // 将修改后的新的值转成 Java 对象
        T entity = this.rowsToEntity(newValues, updateRowsEventData.getTableId());
        return entity;
    }

    protected T deleteEvent(EventData eventData) {
        DeleteRowsEventData deleteRowsEventData = (DeleteRowsEventData) eventData;
        List<Serializable[]> rows = deleteRowsEventData.getRows();
        T entity = this.rowsToEntity(rows, deleteRowsEventData.getTableId());
        return entity;
    }


    protected T rowsToEntity(List<Serializable[]> rows, Long tableId) {
        String tableFullName = tableFullNameMap.get(tableId);
        String[] tableNameArr = tableFullName.split("\\.");

        // 获得当前 row 的数据库中对应的字段名称
        String[] columnNames = sqlSchemaColumnService.columnsByTable(tableNameArr[0], tableNameArr[1]);
        JSONObject beanJSON = new JSONObject();
        for (Serializable[] row : rows) {
            for (int i = 0; i < row.length; i++) {
                beanJSON.put(columnNames[i], row[i]);
            }
        }
        T entity = this.jsonToBean(beanJSON);
        return entity;
    }

    protected T jsonToBean(JSONObject beanJSON) {
        T entity = JSON.toJavaObject(beanJSON, this.entityClass());
        return entity;
    }


    /**
     * 获取调用方法实现类中泛型的具体类对象
     *
     * @return
     */
    protected Class<T> entityClass() {
        // 当前调用方法的 Impl实现类的父类的类型
        ParameterizedType superclass = (ParameterizedType) this.getClass().getGenericSuperclass();
        // 当前调用方法的 Impl实现类的泛型的类型,实现类必须带泛型,否则报错
        Type[] type = superclass.getActualTypeArguments();
        Class clazz = (Class) type[0];
        return clazz;
    }

    /**
     * 获取实体类映射的数据库表名称
     *
     * @param entityClazz
     * @return
     */
    protected String entityTableName(Class<?> entityClazz) {
        boolean isAnno = entityClazz.isAnnotationPresent(TableName.class);
        if (isAnno) {
            TableName annotation = entityClazz.getAnnotation(TableName.class);
            // 表名 @TableName(value = "person_information")
            return annotation.value();
        }
        throw new IRuntimeException("操作不允许", "ERROR");
    }
}